package com.jhhe.mr.mysort;

import com.jhhe.mr.sort.SortDriver;
import com.jhhe.mr.sort.SortMapper;
import com.jhhe.mr.sort.SortReducer;
import com.jhhe.mr.sort.SortSpeakBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
public class MySortDriver {
    // 思路：通过全排序，Reducer遍历values的时候加上序号(通过全局变量来记录序号)，并打印。

    public static Long countN = 0L;

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1 配置config
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "MySortDriver");

        //2 设置jar包本地路径
        job.setJarByClass(MySortDriver.class);

        //3 使用的mapper和reducer
        job.setMapperClass(MySortMapper.class);
        job.setReducerClass(MySortReducer.class);

        //3 map的输出kv类型
        job.setMapOutputKeyClass(MySortBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        //4 设置reduce输出
        job.setOutputKeyClass(MySortBean.class);
        job.setOutputValueClass(NullWritable.class);

        //5 读取的数据路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //6 提交任务
        final boolean flag = job.waitForCompletion(true);
        System.exit(flag ? 0 : 1);
    }
}
